package com.edu.yx.app.App_04_DwsApp.Dws_05_Exam;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.edu.yx.app.APP_01_BaseApp.BaseApp;
import com.edu.yx.bean.ExamExamBean;
import com.edu.yx.bean.ExamExamResultBean;
import com.edu.yx.util.TempUtil;
import com.edu.yx.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;

import static com.edu.yx.common.Constant.*;

public class Dws_01_Exam_Exam_Count extends BaseApp {

    public static void main(String[] args) {
        new Dws_01_Exam_Exam_Count().init(
                4051,
                2,
                "Dws_01_Exam_Exam_Count",
                TOPIC_DWD_EXAM_PAPER_DETAIL
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //1.封装pojo
        SingleOutputStreamOperator<ExamExamBean> parseToPojoStream = parseToPojo(stream);

        //2.开窗聚合
        SingleOutputStreamOperator<ExamExamBean> windowAndAggStream = windowAndAgg(parseToPojoStream);

        //3.二次封装
        SingleOutputStreamOperator<ExamExamResultBean> toPojoAgainStream = parseToPojoAgain(windowAndAggStream);

        //4.写入doris
        writeDoris(toPojoAgainStream);



    }

    private void writeDoris(SingleOutputStreamOperator<ExamExamResultBean> resultStream) {
        resultStream
                .map(bean ->{
                    //配置json可以将驼峰转换成带下划线，然后方便导入数据到doris
                    SerializeConfig config = new SerializeConfig();
                    config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;

                    String json = JSON.toJSONString(bean, config);
//                    System.out.println(json);
                    return json;
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_exam_exam_count_window"));

    }

    private SingleOutputStreamOperator<ExamExamResultBean> parseToPojoAgain(SingleOutputStreamOperator<ExamExamBean> windowAndAggStream) {

         return windowAndAggStream
                .map(bean->{

                    //分子
                    Long score = bean.getScore();
                    Long durationSec = bean.getDurationSec();

                    //分母
                    Long personCt = bean.getPersonCt();

                    Double avgScore = (score * 100.0 / personCt) / 100;
                    Double avgDurationSec = (durationSec * 100.0 / personCt) / 100;

                    return new ExamExamResultBean(
                            bean.getStt(),
                            bean.getEdt(),
                            bean.getPaperId(),
                            bean.getCurDate(),
                            bean.getPersonCt(),
                            avgScore,
                            avgDurationSec
                    );
                });

    }

    private SingleOutputStreamOperator<ExamExamBean> windowAndAgg(SingleOutputStreamOperator<ExamExamBean> parseToPojoStream) {

        return parseToPojoStream
                .assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<ExamExamBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean,ts)->bean.getTs())
                )
                .keyBy(bean->bean.getPaperId())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<ExamExamBean>() {

                            @Override
                            public ExamExamBean reduce(ExamExamBean value1,
                                                       ExamExamBean value2) throws Exception {

                                value1.getPersonList().addAll(value2.getPersonList());
                                value1.setScore(value1.getScore()+ value2.getScore());
                                value1.setDurationSec(value1.getDurationSec() + value2.getDurationSec());

                                return value1;
                            }
                        },
                        new ProcessWindowFunction<ExamExamBean, ExamExamBean, String, TimeWindow>() {
                            @Override
                            public void process(String s,
                                                Context context,
                                                Iterable<ExamExamBean> elements,
                                                Collector<ExamExamBean> out) throws Exception {

                                ExamExamBean bean = elements.iterator().next();

                                bean.setStt(TempUtil.toDateTime(context.window().getStart()));
                                bean.setEdt(TempUtil.toDateTime(context.window().getEnd()));
                                bean.setCurDate(TempUtil.toDate(System.currentTimeMillis()));

                                bean.setPersonCt((long) bean.getPersonList().size());
                                out.collect(bean);

                            }
                        }
                );


    }

    private SingleOutputStreamOperator<ExamExamBean> parseToPojo(DataStreamSource<String> stream) {

        //todo 考试人数，是基于试卷粒度计算-> 某个试卷，在某个时间窗口内，有多少人考了
        return stream
                .map(strjson-> {

                    JSONObject jsonObject = JSON.parseObject(strjson);

                    //拿数据准备封装
                    String id = jsonObject.getString("id");
                    String userId = jsonObject.getString("user_id");
                    Long score = jsonObject.getLong("score");
                    Long durationSec = jsonObject.getLong("duration_sec");
                    Long ts = jsonObject.getLong("ts") * 1000L;


                    return new ExamExamBean(
                            "",
                            "",
                            id,
                            "",
                            new ArrayList<>(Collections.singleton(userId)),
                            0L,
                            score,
                            durationSec,
                            ts
                    );
                });



    }
}
